/*-------------------------------------------------------------------------
 *
 * basebackup.c
 *	  进行基础备份并将其流式传输到备用的代码
 *
 * Portions Copyright (c) 2010-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *	  src/backend/backup/basebackup.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include <sys/stat.h>
#include <unistd.h>
#include <time.h>

#include "access/xlog_internal.h"
#include "backup/backup_manifest.h"
#include "backup/basebackup.h"
#include "backup/basebackup_sink.h"
#include "backup/basebackup_target.h"
#include "commands/defrem.h"
#include "common/compression.h"
#include "common/file_perm.h"
#include "lib/stringinfo.h"
#include "miscadmin.h"
#include "nodes/pg_list.h"
#include "pgstat.h"
#include "pgtar.h"
#include "port.h"
#include "postmaster/syslogger.h"
#include "replication/walsender.h"
#include "replication/walsender_private.h"
#include "storage/bufpage.h"
#include "storage/checksum.h"
#include "storage/dsm_impl.h"
#include "storage/fd.h"
#include "storage/ipc.h"
#include "storage/reinit.h"
#include "utils/builtins.h"
#include "utils/ps_status.h"
#include "utils/relcache.h"
#include "utils/resowner.h"
#include "utils/timestamp.h"

/*
 * 我们希望在一条 CopyData 消息中发送多少数据？请注意
 * 这也可能导致以这个大小读取底层文件。
 *
 * 注意：缓冲区大小必须是系统块大小的倍数，
 * 所以如果它大于我们的偏好，请使用该值。
 */
#define SINK_BUFFER_LENGTH			Max(32768, BLCKSZ)

typedef struct
{
	const char *label;
	bool		progress;
	bool		fastcheckpoint;
	bool		nowait;
	bool		includewal;
	uint32		maxrate;
	bool		sendtblspcmapfile;
	bool		send_to_client;
	bool		use_copytblspc;
	BaseBackupTargetHandle *target_handle;
	backup_manifest_option manifest;
	pg_compress_algorithm compression;
	pg_compress_specification compression_specification;
	pg_checksum_type manifest_checksum_type;
} basebackup_options;

static int64 fc_sendTablespace(bbsink *fc_sink, char *fc_path, char *fc_oid, bool fc_sizeonly,
							struct backup_manifest_info *fc_manifest);
static int64 fc_sendDir(bbsink *fc_sink, const char *fc_path, int fc_basepathlen, bool fc_sizeonly,
					 List *fc_tablespaces, bool fc_sendtblspclinks,
					 backup_manifest_info *fc_manifest, const char *fc_spcoid);
static bool fc_sendFile(bbsink *fc_sink, const char *fc_readfilename, const char *fc_tarfilename,
					 struct stat *fc_statbuf, bool fc_missing_ok, Oid fc_dboid,
					 backup_manifest_info *fc_manifest, const char *fc_spcoid);
static void fc_sendFileWithContent(bbsink *fc_sink, const char *fc_filename,
								const char *fc_content,
								backup_manifest_info *fc_manifest);
static int64 fc__tarWriteHeader(bbsink *fc_sink, const char *fc_filename,
							 const char *fc_linktarget, struct stat *fc_statbuf,
							 bool fc_sizeonly);
static void fc__tarWritePadding(bbsink *fc_sink, int fc_len);
static void fc_convert_link_to_directory(const char *fc_pathbuf, struct stat *fc_statbuf);
static void fc_perform_base_backup(basebackup_options *fc_opt, bbsink *fc_sink);
static void fc_parse_basebackup_options(List *fc_options, basebackup_options *fc_opt);
static int	fc_compareWalFileNames(const ListCell *fc_a, const ListCell *fc_b);
static bool fc_is_checksummed_file(const char *fc_fullpath, const char *fc_filename);
static int	fc_basebackup_read_file(int fc_fd, char *fc_buf, size_t fc_nbytes, off_t fc_offset,
								 const char *fc_filename, bool fc_partial_read_ok);

/* 当前正在进行的备份是在恢复模式下启动的吗？ */
static bool backup_started_in_recovery = false;

/* 在基础备份期间的校验和失败总数。 */
static long long int total_checksum_failures;

/* 不要验证校验和。 */
static bool noverify_checksums = false;

/*
 * 排除列表中一个元素部分的定义，用于路径部分
 * 的校验和验证或基础备份。 "name" 是要检查排除的文件
 * 或路径的名称。如果 "match_prefix" 为真，则任何与名称匹配的前缀项
 * 都将被排除。
 */
struct exclude_list_item
{
	const char *name;
	bool		match_prefix;
};

/*
 * 这些目录的内容在服务器启动时被移除或重新创建，
 * 因此它们不包含在备份中。目录本身被保留，
 * 并以空目录的形式包含，以保持访问权限。
 *
 * 注意：该列表应与 pg_rewind 的 filemap.c 中的过滤器列表保持同步。
 */
static const char *const excludeDirContents[] =
{
	/*
	 * 跳过临时统计文件。必须跳过 PG_STAT_TMP_DIR，
	 * 因为像 pg_stat_statements 这样的扩展在其中存储数据。
	 */
	PG_STAT_TMP_DIR,

#ifndef FDD
	/*
	 * 即使打算恢复到另一个主节点，备份此目录的内容
	 * 通常也没有用。有关更详细的描述，请参见 backup.sgml。
	 */
	"pg_replslot",
#endif

	/* 启动时移除内容，请参见 dsm_cleanup_for_mmap()。 */
	PG_DYNSHMEM_DIR,

	/* 启动时移除内容，请参见 AsyncShmemInit()。 */
	"pg_notify",

	/*
	 * 旧内容被加载以便进行调试，但对
	 * 正常操作不是必需的，请参见 SerialInit()。
	 */
	"pg_serial",

	/* 启动时移除内容，请参见 DeleteAllExportedSnapshotFiles()。 */
	"pg_snapshots",

	/* 启动时内容为零，请参见 StartupSUBTRANS()。 */
	"pg_subtrans",

	/* 列表结束 */
	NULL
};

/*
 * 排除在备份中的文件列表。
 */
static const struct exclude_list_item excludeFiles[] =
{
	/* 跳过自动配置临时文件。 */
	{PG_AUTOCONF_FILENAME ".tmp", false},

	/* 跳过当前日志文件临时文件 */
	{LOG_METAINFO_DATAFILE_TMP, false},

	/*
	 * 跳过关系缓存，因为它在启动时会重建。这包括
	 * 临时文件。
	 */
	{RELCACHE_INIT_FILENAME, true},

	/*
	 * backup_label 和 tablespace_map 在能够进行在线备份的
	 * 运行集群中不应存在，但为了以防万一将它们排除。
	 */
	{BACKUP_LABEL_FILE, false},
	{TABLESPACE_MAP, false},

	/*
	 * 如果存在 backup_manifest，则它属于用于
	 * 启动此服务器的备份。对于此备份，它 *不是* 正确的。
	 * 我们的 backup_manifest 是在备份中单独注入的，
	 * 如果用户需要它的话。
	 */
	{"backup_manifest", false},

	{"postmaster.pid", false},
	{"postmaster.opts", false},

	/* 列表结束 */
	{NULL, false}
};

/*
 * 排除在校验和验证中的文件列表。
 *
 * 注意：该列表应与 pg_checksums.c 中包含的内容保持同步。
 */
static const struct exclude_list_item noChecksumFiles[] = {
	{"pg_control", false},
	{"pg_filenode.map", false},
	{"pg_internal.init", true},
	{"PG_VERSION", false},
#ifdef EXEC_BACKEND
	{"config_exec_params", true},
#endif
	{NULL, false}
};

/*
 * 实际上对指定的表空间进行基础备份。
 *
 * 这主要是为了避免来自更笨版本的 gcc 的 "变量可能被
 * longjmp 覆盖" 的投诉。
 */
static void fc_perform_base_backup(basebackup_options *fc_opt, bbsink *fc_sink)
{
	bbsink_state fc_state;
	XLogRecPtr	fc_endptr;
	TimeLineID	fc_endtli;
	StringInfo	fc_labelfile;
	StringInfo	fc_tblspc_map_file;
	backup_manifest_info fc_manifest;

	/* 初始备份状态，就我们现在所知。 */
	fc_state.tablespaces = NIL;
	fc_state.tablespace_num = 0;
	fc_state.bytes_done = 0;
	fc_state.bytes_total = 0;
	fc_state.bytes_total_is_valid = false;

	/* 我们将使用 BufFile，因此需要一个 ResourceOwner */
	Assert(CurrentResourceOwner == NULL);
	CurrentResourceOwner = ResourceOwnerCreate(NULL, "base backup");

	backup_started_in_recovery = RecoveryInProgress();

	fc_labelfile = makeStringInfo();
	fc_tblspc_map_file = makeStringInfo();
	InitializeBackupManifest(&fc_manifest, fc_opt->manifest,
							 fc_opt->manifest_checksum_type);

	total_checksum_failures = 0;

	basebackup_progress_wait_checkpoint();
	fc_state.startptr = do_pg_backup_start(fc_opt->label, fc_opt->fastcheckpoint,
										&fc_state.starttli,
										fc_labelfile, &fc_state.tablespaces,
										fc_tblspc_map_file);

	/*
	 * 一旦调用 do_pg_backup_start，确保任何失败都导致我们中止备份，以防止我们“泄漏”备份计数器。出于这个原因，do_pg_backup_start() 和 do_pg_backup_stop() 结束之间的 *所有* 功能应该在错误清理块内!
	 */

	PG_ENSURE_ERROR_CLEANUP(do_pg_abort_backup, BoolGetDatum(false));
	{
		ListCell   *fc_lc;
		tablespaceinfo *fc_ti;

		/* 在末尾为基础目录添加一个节点 */
		fc_ti = palloc0(sizeof(tablespaceinfo));
		fc_ti->size = -1;
		fc_state.tablespaces = lappend(fc_state.tablespaces, fc_ti);

		/*
		 * 通过将每个 tablespace 的大小相加来计算总备份大小
		 */
		if (fc_opt->progress)
		{
			basebackup_progress_estimate_backup_size();

			foreach(fc_lc, fc_state.tablespaces)
			{
				tablespaceinfo *fc_tmp = (tablespaceinfo *) lfirst(fc_lc);

				if (fc_tmp->path == NULL)
					fc_tmp->size = fc_sendDir(fc_sink, ".", 1, true, fc_state.tablespaces,
										true, NULL, NULL);
				else
					fc_tmp->size = fc_sendTablespace(fc_sink, fc_tmp->path, fc_tmp->oid, true,
											   NULL);
				fc_state.bytes_total += fc_tmp->size;
			}
			fc_state.bytes_total_is_valid = true;
		}

		/* 通知 basebackup sink 备份开始 */
		bbsink_begin_backup(fc_sink, &fc_state, SINK_BUFFER_LENGTH);

		/* 一个一个地发送我们的 tablespaces */
		foreach(fc_lc, fc_state.tablespaces)
		{
			tablespaceinfo *fc_ti = (tablespaceinfo *) lfirst(fc_lc);

			if (fc_ti->path == NULL)
			{
				struct stat fc_statbuf;
				bool		fc_sendtblspclinks = true;

				bbsink_begin_archive(fc_sink, "base.tar");

				/* 在主 tar 中，首先包含 backup_label... */
				fc_sendFileWithContent(fc_sink, BACKUP_LABEL_FILE, fc_labelfile->data,
									&fc_manifest);

				/* 然后是 tablespace_map 文件（如果需要的话）... */
				if (fc_opt->sendtblspcmapfile)
				{
					fc_sendFileWithContent(fc_sink, TABLESPACE_MAP, fc_tblspc_map_file->data,
										&fc_manifest);
					fc_sendtblspclinks = false;
				}

				/* 接着是大部分文件... */
				fc_sendDir(fc_sink, ".", 1, false, fc_state.tablespaces,
						fc_sendtblspclinks, &fc_manifest, NULL);

				/* ... 然后是 pg_control 在最后。 */
				if (lstat(XLOG_CONTROL_FILE, &fc_statbuf) != 0)
					ereport(ERROR,
							(errcode_for_file_access(),
							 errmsg("could not stat file \"%s\": %m",
									XLOG_CONTROL_FILE)));
				fc_sendFile(fc_sink, XLOG_CONTROL_FILE, XLOG_CONTROL_FILE, &fc_statbuf,
						 false, InvalidOid, &fc_manifest, NULL);
			}
			else
			{
				char	   *fc_archive_name = psprintf("%s.tar", fc_ti->oid);

				bbsink_begin_archive(fc_sink, fc_archive_name);

				fc_sendTablespace(fc_sink, fc_ti->path, fc_ti->oid, false, &fc_manifest);
			}

			/*
			 * 如果我们包括 WAL，并且这是主数据目录，我们
			 * 不将其视为 tablespace 的结束。相反，我们将
			 * 包括下面的 xlog 文件，并在之后停止。这是安全的
			 *，因为主数据目录总是最后发送。
			 */
			if (fc_opt->includewal && fc_ti->path == NULL)
			{
				Assert(lnext(fc_state.tablespaces, fc_lc) == NULL);
			}
			else
			{
				/* 正确终止 tarfile。 */
				StaticAssertStmt(2 * TAR_BLOCK_SIZE <= BLCKSZ,
								 "BLCKSZ too small for 2 tar blocks");
				memset(fc_sink->bbs_buffer, 0, 2 * TAR_BLOCK_SIZE);
				bbsink_archive_contents(fc_sink, 2 * TAR_BLOCK_SIZE);

				/* 好的，这就是归档的结束。 */
				bbsink_end_archive(fc_sink);
			}
		}

		basebackup_progress_wait_wal_archive(&fc_state);
		fc_endptr = do_pg_backup_stop(fc_labelfile->data, !fc_opt->nowait, &fc_endtli);
	}
	PG_END_ENSURE_ERROR_CLEANUP(do_pg_abort_backup, BoolGetDatum(false));


	if (fc_opt->includewal)
	{
		/*
		 * 我们把最后的 tar 文件“打开”，现在可以将
		 * 所需的 WAL 文件追加到其中。
		 */
		char		fc_pathbuf[MAXPGPATH];
		XLogSegNo	fc_segno;
		XLogSegNo	fc_startsegno;
		XLogSegNo	fc_endsegno;
		struct stat fc_statbuf;
		List	   *fc_historyFileList = NIL;
		List	   *fc_walFileList = NIL;
		char		fc_firstoff[MAXFNAMELEN];
		char		fc_lastoff[MAXFNAMELEN];
		DIR		   *fc_dir;
		struct dirent *fc_de;
		ListCell   *fc_lc;
		TimeLineID	fc_tli;

		basebackup_progress_transfer_wal();

		/*
		 * 我不想在这里担心时间线，所以扫描 pg_wal 并
		 * 包含所有在 'startptr' 和 'endptr' 范围内的 WAL 文件，
		 * 无论文件的时间线是什么。如果有一些属于不在
		 * 该服务器历史中的时间线的多余 WAL 文件，它们也会被包括。正常情况下
		 * 不应该有这样的文件，但如果有，包含它们也不会造成太大 harm。
		 */
		XLByteToSeg(fc_state.startptr, fc_startsegno, wal_segment_size);
		XLogFileName(fc_firstoff, fc_state.starttli, fc_startsegno, wal_segment_size);
		XLByteToPrevSeg(fc_endptr, fc_endsegno, wal_segment_size);
		XLogFileName(fc_lastoff, fc_endtli, fc_endsegno, wal_segment_size);

		fc_dir = AllocateDir("pg_wal");
		while ((fc_de = ReadDir(fc_dir, "pg_wal")) != NULL)
		{
			/* 看起来像 WAL 段吗，并且在范围内吗？ */
			if (IsXLogFileName(fc_de->d_name) &&
				strcmp(fc_de->d_name + 8, fc_firstoff + 8) >= 0 &&
				strcmp(fc_de->d_name + 8, fc_lastoff + 8) <= 0)
			{
				fc_walFileList = lappend(fc_walFileList, pstrdup(fc_de->d_name));
			}
			/* 看起来像时间线历史文件吗？ */
			else if (IsTLHistoryFileName(fc_de->d_name))
			{
				fc_historyFileList = lappend(fc_historyFileList, pstrdup(fc_de->d_name));
			}
		}
		FreeDir(fc_dir);

		/*
		 * 在我们继续之前，检查我们需要的 WAL 段是否
		 * 被移除。
		 */
		CheckXLogRemoved(fc_startsegno, fc_state.starttli);

		/*
		 * 排序 WAL 文件名。我们希望按从
		 * 最旧到最新的顺序发送文件，以减少在我们有机会
		 * 发送它之前文件被回收的可能性。
		 */
		list_sort(fc_walFileList, fc_compareWalFileNames);

		/*
		 * pg_wal 目录中必须至少有一个 xlog 文件，因为
		 * 我们在进行备份并包含 xlog。
		 */
		if (fc_walFileList == NIL)
			ereport(ERROR,
					(errmsg("could not find any WAL files")));

		/*
		 * 合理性检查：第一和最后的段应该覆盖 startptr 和
		 * endptr，之间没有间隙。
		 */
		XLogFromFileName((char *) linitial(fc_walFileList),
						 &fc_tli, &fc_segno, wal_segment_size);
		if (fc_segno != fc_startsegno)
		{
			char		fc_startfname[MAXFNAMELEN];

			XLogFileName(fc_startfname, fc_state.starttli, fc_startsegno,
						 wal_segment_size);
			ereport(ERROR,
					(errmsg("could not find WAL file \"%s\"", fc_startfname)));
		}
		foreach(fc_lc, fc_walFileList)
		{
			char	   *fc_walFileName = (char *) lfirst(fc_lc);
			XLogSegNo	fc_currsegno = fc_segno;
			XLogSegNo	fc_nextsegno = fc_segno + 1;

			XLogFromFileName(fc_walFileName, &fc_tli, &fc_segno, wal_segment_size);
			if (!(fc_nextsegno == fc_segno || fc_currsegno == fc_segno))
			{
				char		fc_nextfname[MAXFNAMELEN];

				XLogFileName(fc_nextfname, fc_tli, fc_nextsegno, wal_segment_size);
				ereport(ERROR,
						(errmsg("could not find WAL file \"%s\"", fc_nextfname)));
			}
		}
		if (fc_segno != fc_endsegno)
		{
			char		fc_endfname[MAXFNAMELEN];

			XLogFileName(fc_endfname, fc_endtli, fc_endsegno, wal_segment_size);
			ereport(ERROR,
					(errmsg("could not find WAL file \"%s\"", fc_endfname)));
		}

		/* 好的，我们有我们需要的所有东西。发送 WAL 文件。 */
		foreach(fc_lc, fc_walFileList)
		{
			char	   *fc_walFileName = (char *) lfirst(fc_lc);
			int			fc_fd;
			size_t		fc_cnt;
			pgoff_t		fc_len = 0;

			snprintf(fc_pathbuf, MAXPGPATH, XLOGDIR "/%s", fc_walFileName);
			XLogFromFileName(fc_walFileName, &fc_tli, &fc_segno, wal_segment_size);

			fc_fd = OpenTransientFile(fc_pathbuf, O_RDONLY | PG_BINARY);
			if (fc_fd < 0)
			{
				int			fc_save_errno = errno;

				/*
				 * 这最有可能的原因是文件已经被检查点移除，
				 * 所以检查一下以获得更好的错误消息。
				 */
				CheckXLogRemoved(fc_segno, fc_tli);

				errno = fc_save_errno;
				ereport(ERROR,
						(errcode_for_file_access(),
						 errmsg("could not open file \"%s\": %m", fc_pathbuf)));
			}

			if (fstat(fc_fd, &fc_statbuf) != 0)
				ereport(ERROR,
						(errcode_for_file_access(),
						 errmsg("could not stat file \"%s\": %m",
								fc_pathbuf)));
			if (fc_statbuf.st_size != wal_segment_size)
			{
				CheckXLogRemoved(fc_segno, fc_tli);
				ereport(ERROR,
						(errcode_for_file_access(),
						 errmsg("unexpected WAL file size \"%s\"", fc_walFileName)));
			}

			/* 发送 WAL 文件本身 */
			fc__tarWriteHeader(fc_sink, fc_pathbuf, NULL, &fc_statbuf, false);

			while ((fc_cnt = fc_basebackup_read_file(fc_fd, fc_sink->bbs_buffer,
											   Min(fc_sink->bbs_buffer_length,
												   wal_segment_size - fc_len),
											   fc_len, fc_pathbuf, true)) > 0)
			{
				CheckXLogRemoved(fc_segno, fc_tli);
				bbsink_archive_contents(fc_sink, fc_cnt);

				fc_len += fc_cnt;

				if (fc_len == wal_segment_size)
					break;
			}

			if (fc_len != wal_segment_size)
			{
				CheckXLogRemoved(fc_segno, fc_tli);
				ereport(ERROR,
						(errcode_for_file_access(),
						 errmsg("unexpected WAL file size \"%s\"", fc_walFileName)));
			}

			/*
			 * wal_segment_size 是 TAR_BLOCK_SIZE 的整数倍，因此不需要
			 * 填充。
			 */
			Assert(wal_segment_size % TAR_BLOCK_SIZE == 0);

			CloseTransientFile(fc_fd);

			/*
			 * 将文件标记为已归档，否则在新的节点提升后，
			 * 文件可能会再次被归档。这与
			 * walreceiver.c 在完整段后总是执行 XLogArchiveForceDone() 一致。
			 */
			StatusFilePath(fc_pathbuf, fc_walFileName, ".done");
			fc_sendFileWithContent(fc_sink, fc_pathbuf, "", &fc_manifest);
		}

		/*
		 * 也发送时间线历史文件。恢复时仅需要最新的时间线历史
		 * 文件，甚至只有当发生时间线切换并且第一个WAL段包含
		 * 检查点记录，或者我们从备用服务器进行基础备份
		 * 时目标时间线在备份过程中发生更改时才需要。
		 * 但它们的体积小且在调试时非常有用，因此
		 * 最好始终包含所有文件。
		 */
		foreach(fc_lc, fc_historyFileList)
		{
			char	   *fc_fname = lfirst(fc_lc);

			snprintf(fc_pathbuf, MAXPGPATH, XLOGDIR "/%s", fc_fname);

			if (lstat(fc_pathbuf, &fc_statbuf) != 0)
				ereport(ERROR,
						(errcode_for_file_access(),
						 errmsg("could not stat file \"%s\": %m", fc_pathbuf)));

			fc_sendFile(fc_sink, fc_pathbuf, fc_pathbuf, &fc_statbuf, false, InvalidOid,
					 &fc_manifest, NULL);

			/* 无条件标记文件为已归档 */
			StatusFilePath(fc_pathbuf, fc_fname, ".done");
			fc_sendFileWithContent(fc_sink, fc_pathbuf, "", &fc_manifest);
		}

		/* 正确终止tar文件。 */
		StaticAssertStmt(2 * TAR_BLOCK_SIZE <= BLCKSZ,
						 "BLCKSZ too small for 2 tar blocks");
		memset(fc_sink->bbs_buffer, 0, 2 * TAR_BLOCK_SIZE);
		bbsink_archive_contents(fc_sink, 2 * TAR_BLOCK_SIZE);

		/* 好的，这就是归档的结束。 */
		bbsink_end_archive(fc_sink);
	}

	AddWALInfoToBackupManifest(&fc_manifest, fc_state.startptr, fc_state.starttli,
							   fc_endptr, fc_endtli);

	SendBackupManifest(&fc_manifest, fc_sink);

	bbsink_end_backup(fc_sink, fc_endptr, fc_endtli);

	if (total_checksum_failures)
	{
		if (total_checksum_failures > 1)
			ereport(WARNING,
					(errmsg_plural("%lld total checksum verification failure",
								   "%lld total checksum verification failures",
								   total_checksum_failures,
								   total_checksum_failures)));

		ereport(ERROR,
				(errcode(ERRCODE_DATA_CORRUPTED),
				 errmsg("checksum verification failure during base backup")));
	}

	/*
	 * 确保在资源所有者之前释放清单，因为清单
	 * 使用可能依赖于资源所有者（如
	 * OpenSSL）的cryptohash上下文。
	 */
	FreeBackupManifest(&fc_manifest);

	/* 清理我们创建的资源所有者 */
	WalSndResourceCleanup(true);

	basebackup_progress_done();
}

/*
 * list_sort比较函数，用于比较WAL段
 * 文件名的日志/段部分，忽略时间线部分。
 */
static int fc_compareWalFileNames(const ListCell *fc_a, const ListCell *fc_b)
{
	char	   *fc_fna = (char *) lfirst(fc_a);
	char	   *fc_fnb = (char *) lfirst(fc_b);

	return strcmp(fc_fna + 8, fc_fnb + 8);
}

/*
 * 解析解析器传递下来的基础备份选项
 */
static void fc_parse_basebackup_options(List *fc_options, basebackup_options *fc_opt)
{
	ListCell   *fc_lopt;
	bool		fc_o_label = false;
	bool		fc_o_progress = false;
	bool		fc_o_checkpoint = false;
	bool		fc_o_nowait = false;
	bool		fc_o_wal = false;
	bool		fc_o_maxrate = false;
	bool		fc_o_tablespace_map = false;
	bool		fc_o_noverify_checksums = false;
	bool		fc_o_manifest = false;
	bool		fc_o_manifest_checksums = false;
	bool		fc_o_target = false;
	bool		fc_o_target_detail = false;
	char	   *fc_target_str = NULL;
	char	   *fc_target_detail_str = NULL;
	bool		fc_o_compression = false;
	bool		fc_o_compression_detail = false;
	char	   *fc_compression_detail_str = NULL;

	MemSet(fc_opt, 0, sizeof(*fc_opt));
	fc_opt->manifest = MANIFEST_OPTION_NO;
	fc_opt->manifest_checksum_type = CHECKSUM_TYPE_CRC32C;
	fc_opt->compression = PG_COMPRESSION_NONE;
	fc_opt->compression_specification.algorithm = PG_COMPRESSION_NONE;

	foreach(fc_lopt, fc_options)
	{
		DefElem    *fc_defel = (DefElem *) lfirst(fc_lopt);

		if (strcmp(fc_defel->defname, "label") == 0)
		{
			if (fc_o_label)
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("duplicate option \"%s\"", fc_defel->defname)));
			fc_opt->label = defGetString(fc_defel);
			fc_o_label = true;
		}
		else if (strcmp(fc_defel->defname, "progress") == 0)
		{
			if (fc_o_progress)
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("duplicate option \"%s\"", fc_defel->defname)));
			fc_opt->progress = defGetBoolean(fc_defel);
			fc_o_progress = true;
		}
		else if (strcmp(fc_defel->defname, "checkpoint") == 0)
		{
			char	   *fc_optval = defGetString(fc_defel);

			if (fc_o_checkpoint)
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("duplicate option \"%s\"", fc_defel->defname)));
			if (pg_strcasecmp(fc_optval, "fast") == 0)
				fc_opt->fastcheckpoint = true;
			else if (pg_strcasecmp(fc_optval, "spread") == 0)
				fc_opt->fastcheckpoint = false;
			else
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("unrecognized checkpoint type: \"%s\"",
								fc_optval)));
			fc_o_checkpoint = true;
		}
		else if (strcmp(fc_defel->defname, "wait") == 0)
		{
			if (fc_o_nowait)
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("duplicate option \"%s\"", fc_defel->defname)));
			fc_opt->nowait = !defGetBoolean(fc_defel);
			fc_o_nowait = true;
		}
		else if (strcmp(fc_defel->defname, "wal") == 0)
		{
			if (fc_o_wal)
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("duplicate option \"%s\"", fc_defel->defname)));
			fc_opt->includewal = defGetBoolean(fc_defel);
			fc_o_wal = true;
		}
		else if (strcmp(fc_defel->defname, "max_rate") == 0)
		{
			int64		fc_maxrate;

			if (fc_o_maxrate)
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("duplicate option \"%s\"", fc_defel->defname)));

			fc_maxrate = defGetInt64(fc_defel);
			if (fc_maxrate < MAX_RATE_LOWER || fc_maxrate > MAX_RATE_UPPER)
				ereport(ERROR,
						(errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
						 errmsg("%d is outside the valid range for parameter \"%s\" (%d .. %d)",
								(int) fc_maxrate, "MAX_RATE", MAX_RATE_LOWER, MAX_RATE_UPPER)));

			fc_opt->maxrate = (uint32) fc_maxrate;
			fc_o_maxrate = true;
		}
		else if (strcmp(fc_defel->defname, "tablespace_map") == 0)
		{
			if (fc_o_tablespace_map)
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("duplicate option \"%s\"", fc_defel->defname)));
			fc_opt->sendtblspcmapfile = defGetBoolean(fc_defel);
			fc_o_tablespace_map = true;
		}
		else if (strcmp(fc_defel->defname, "verify_checksums") == 0)
		{
			if (fc_o_noverify_checksums)
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("duplicate option \"%s\"", fc_defel->defname)));
			noverify_checksums = !defGetBoolean(fc_defel);
			fc_o_noverify_checksums = true;
		}
		else if (strcmp(fc_defel->defname, "manifest") == 0)
		{
			char	   *fc_optval = defGetString(fc_defel);
			bool		fc_manifest_bool;

			if (fc_o_manifest)
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("duplicate option \"%s\"", fc_defel->defname)));
			if (parse_bool(fc_optval, &fc_manifest_bool))
			{
				if (fc_manifest_bool)
					fc_opt->manifest = MANIFEST_OPTION_YES;
				else
					fc_opt->manifest = MANIFEST_OPTION_NO;
			}
			else if (pg_strcasecmp(fc_optval, "force-encode") == 0)
				fc_opt->manifest = MANIFEST_OPTION_FORCE_ENCODE;
			else
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("unrecognized manifest option: \"%s\"",
								fc_optval)));
			fc_o_manifest = true;
		}
		else if (strcmp(fc_defel->defname, "manifest_checksums") == 0)
		{
			char	   *fc_optval = defGetString(fc_defel);

			if (fc_o_manifest_checksums)
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("duplicate option \"%s\"", fc_defel->defname)));
			if (!pg_checksum_parse_type(fc_optval,
										&fc_opt->manifest_checksum_type))
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("unrecognized checksum algorithm: \"%s\"",
								fc_optval)));
			fc_o_manifest_checksums = true;
		}
		else if (strcmp(fc_defel->defname, "target") == 0)
		{
			if (fc_o_target)
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("duplicate option \"%s\"", fc_defel->defname)));
			fc_target_str = defGetString(fc_defel);
			fc_o_target = true;
		}
		else if (strcmp(fc_defel->defname, "target_detail") == 0)
		{
			char	   *fc_optval = defGetString(fc_defel);

			if (fc_o_target_detail)
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("duplicate option \"%s\"", fc_defel->defname)));
			fc_target_detail_str = fc_optval;
			fc_o_target_detail = true;
		}
		else if (strcmp(fc_defel->defname, "compression") == 0)
		{
			char	   *fc_optval = defGetString(fc_defel);

			if (fc_o_compression)
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("duplicate option \"%s\"", fc_defel->defname)));
			if (!parse_compress_algorithm(fc_optval, &fc_opt->compression))
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("unrecognized compression algorithm: \"%s\"",
								fc_optval)));
			fc_o_compression = true;
		}
		else if (strcmp(fc_defel->defname, "compression_detail") == 0)
		{
			if (fc_o_compression_detail)
				ereport(ERROR,
						(errcode(ERRCODE_SYNTAX_ERROR),
						 errmsg("duplicate option \"%s\"", fc_defel->defname)));
			fc_compression_detail_str = defGetString(fc_defel);
			fc_o_compression_detail = true;
		}
		else
			ereport(ERROR,
					(errcode(ERRCODE_SYNTAX_ERROR),
					 errmsg("unrecognized base backup option: \"%s\"",
							fc_defel->defname)));
	}

	if (fc_opt->label == NULL)
		fc_opt->label = "base backup";
	if (fc_opt->manifest == MANIFEST_OPTION_NO)
	{
		if (fc_o_manifest_checksums)
			ereport(ERROR,
					(errcode(ERRCODE_SYNTAX_ERROR),
					 errmsg("manifest checksums require a backup manifest")));
		fc_opt->manifest_checksum_type = CHECKSUM_TYPE_NONE;
	}

	if (fc_target_str == NULL)
	{
		if (fc_target_detail_str != NULL)
			ereport(ERROR,
					(errcode(ERRCODE_SYNTAX_ERROR),
					 errmsg("target detail cannot be used without target")));
		fc_opt->use_copytblspc = true;
		fc_opt->send_to_client = true;
	}
	else if (strcmp(fc_target_str, "client") == 0)
	{
		if (fc_target_detail_str != NULL)
			ereport(ERROR,
					(errcode(ERRCODE_SYNTAX_ERROR),
					 errmsg("target \"%s\" does not accept a target detail",
							fc_target_str)));
		fc_opt->send_to_client = true;
	}
	else
		fc_opt->target_handle =
			BaseBackupGetTargetHandle(fc_target_str, fc_target_detail_str);

	if (fc_o_compression_detail && !fc_o_compression)
		ereport(ERROR,
				(errcode(ERRCODE_SYNTAX_ERROR),
				 errmsg("compression detail cannot be specified unless compression is enabled")));

	if (fc_o_compression)
	{
		char	   *fc_error_detail;

		parse_compress_specification(fc_opt->compression, fc_compression_detail_str,
									 &fc_opt->compression_specification);
		fc_error_detail =
			validate_compress_specification(&fc_opt->compression_specification);
		if (fc_error_detail != NULL)
			ereport(ERROR,
					errcode(ERRCODE_SYNTAX_ERROR),
					errmsg("invalid compression specification: %s",
						   fc_error_detail));
	}
}


/*
 * SendBaseBackup() - 发送完整的基础备份。
 *
 * 该函数将系统置于备份模式，就像pg_backup_start()
 * 所做的那样，因此尽管我们直接从
 * 文件系统读取，绕过缓冲区缓存，备份仍然是一致的。
 */
void SendBaseBackup(BaseBackupCmd *fc_cmd)
{
	basebackup_options fc_opt;
	bbsink	   *fc_sink;
	SessionBackupState fc_status = get_backup_status();

	if (fc_status == SESSION_BACKUP_RUNNING)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("a backup is already in progress in this session")));

	fc_parse_basebackup_options(fc_cmd->options, &fc_opt);

	WalSndSetState(WALSNDSTATE_BACKUP);

	if (update_process_title)
	{
		char		fc_activitymsg[50];

		snprintf(fc_activitymsg, sizeof(fc_activitymsg), "sending backup \"%s\"",
				 fc_opt.label);
		set_ps_display(fc_activitymsg);
	}

	/*
	 * 如果目标特定为“客户端”，则设置将备份
	 * 流式传输到客户端；否则，它被发送到其他地方，不应
	 * 发送到客户端。BaseBackupGetSink的工作是设置
	 * 一个接收端，将备份数据发送到需要的位置。
	 */
	fc_sink = bbsink_copystream_new(fc_opt.send_to_client);
	if (fc_opt.target_handle != NULL)
		fc_sink = BaseBackupGetSink(fc_opt.target_handle, fc_sink);

	/* 设置网络节流，如果客户端请求的话 */
	if (fc_opt.maxrate > 0)
		fc_sink = bbsink_throttle_new(fc_sink, fc_opt.maxrate);

	/* 设置服务器端压缩，如果客户端请求的话 */
	if (fc_opt.compression == PG_COMPRESSION_GZIP)
		fc_sink = bbsink_gzip_new(fc_sink, &fc_opt.compression_specification);
	else if (fc_opt.compression == PG_COMPRESSION_LZ4)
		fc_sink = bbsink_lz4_new(fc_sink, &fc_opt.compression_specification);
	else if (fc_opt.compression == PG_COMPRESSION_ZSTD)
		fc_sink = bbsink_zstd_new(fc_sink, &fc_opt.compression_specification);

	/* 设置进度报告。 */
	fc_sink = bbsink_progress_new(fc_sink, fc_opt.progress);

	/*
	 * 执行基础备份，但确保即使发生错误也
	 * 清理bbsink。
	 */
	PG_TRY();
	{
		fc_perform_base_backup(&fc_opt, fc_sink);
	}
	PG_FINALLY();
	{
		bbsink_cleanup(fc_sink);
	}
	PG_END_TRY();
}

/*
 * 在输出的tar流中注入一个具有给定名称和内容的文件。
 */
static void fc_sendFileWithContent(bbsink *fc_sink, const char *fc_filename, const char *fc_content,
					backup_manifest_info *fc_manifest)
{
	struct stat fc_statbuf;
	int			fc_bytes_done = 0,
				fc_len;
	pg_checksum_context fc_checksum_ctx;

	if (pg_checksum_init(&fc_checksum_ctx, fc_manifest->checksum_type) < 0)
		elog(ERROR, "could not initialize checksum of file \"%s\"",
			 fc_filename);

	fc_len = strlen(fc_content);

	/*
	 * 为我们注入的backup_label文件构造一个stat结构。
	 */
	/* Windows没有uid和gid的概念 */
#ifdef WIN32
	statbuf.st_uid = 0;
	statbuf.st_gid = 0;
#else
	fc_statbuf.st_uid = geteuid();
	fc_statbuf.st_gid = getegid();
#endif
	fc_statbuf.st_mtime = time(NULL);
	fc_statbuf.st_mode = pg_file_create_mode;
	fc_statbuf.st_size = fc_len;

	fc__tarWriteHeader(fc_sink, fc_filename, NULL, &fc_statbuf, false);

	if (pg_checksum_update(&fc_checksum_ctx, (uint8 *) fc_content, fc_len) < 0)
		elog(ERROR, "could not update checksum of file \"%s\"",
			 fc_filename);

	while (fc_bytes_done < fc_len)
	{
		size_t		fc_remaining = fc_len - fc_bytes_done;
		size_t		fc_nbytes = Min(fc_sink->bbs_buffer_length, fc_remaining);

		memcpy(fc_sink->bbs_buffer, fc_content, fc_nbytes);
		bbsink_archive_contents(fc_sink, fc_nbytes);
		fc_bytes_done += fc_nbytes;
		fc_content += fc_nbytes;
	}

	fc__tarWritePadding(fc_sink, fc_len);

	AddFileToBackupManifest(fc_manifest, NULL, fc_filename, fc_len,
							(pg_time_t) fc_statbuf.st_mtime, &fc_checksum_ctx);
}

/*
 * 将'path'指向的表空间目录包含在输出tar
 * 流中。如果'sizeonly'为真，我们将计算总长度并返回
 * 它，而不实际发送任何内容。
 *
 * 仅用于发送辅助表空间，而非PGDATA。
 */
static int64 fc_sendTablespace(bbsink *fc_sink, char *fc_path, char *fc_spcoid, bool fc_sizeonly,
			   backup_manifest_info *fc_manifest)
{
	int64		fc_size;
	char		fc_pathbuf[MAXPGPATH];
	struct stat fc_statbuf;

	/*
	 * 'path'指向表空间位置，但我们只想包含
	 * 属于我们的版本目录。
	 */
	snprintf(fc_pathbuf, sizeof(fc_pathbuf), "%s/%s", fc_path,
			 TABLESPACE_VERSION_DIRECTORY);

	/*
	 * 在tar文件中存储一个目录项，以便我们获得正确的权限。
	 */
	if (lstat(fc_pathbuf, &fc_statbuf) != 0)
	{
		if (errno != ENOENT)
			ereport(ERROR,
					(errcode_for_file_access(),
					 errmsg("could not stat file or directory \"%s\": %m",
							fc_pathbuf)));

		/* 如果在扫描时表空间消失，这不是错误。 */
		return 0;
	}

	fc_size = fc__tarWriteHeader(fc_sink, TABLESPACE_VERSION_DIRECTORY, NULL, &fc_statbuf,
						   fc_sizeonly);

	/* 发送表空间版本目录中的所有文件 */
	fc_size += fc_sendDir(fc_sink, fc_pathbuf, strlen(fc_path), fc_sizeonly, NIL, true, fc_manifest,
					fc_spcoid);

	return fc_size;
}

/*
 * 将给定目录中的所有文件包含在输出的 tar 流中。如果
 * 'sizeonly' 为 true，我们只计算总长度并返回，而不
 * 实际发送任何内容。
 *
 * 在表空间列表中省略任何目录，以避免在它们在 PGDATA
 * 中创建时备份表空间两次。
 *
 * 如果 sendtblspclinks 为 true，我们需要在 tar 文件中
 * 包括符号链接信息。如果不是，我们可以跳过，因为
 * 它将单独在 tablespace_map 文件中发送。
 */
static int64 fc_sendDir(bbsink *fc_sink, const char *fc_path, int fc_basepathlen, bool fc_sizeonly,
		List *fc_tablespaces, bool fc_sendtblspclinks, backup_manifest_info *fc_manifest,
		const char *fc_spcoid)
{
	DIR		   *fc_dir;
	struct dirent *fc_de;
	char		fc_pathbuf[MAXPGPATH * 2];
	struct stat fc_statbuf;
	int64		fc_size = 0;
	const char *fc_lastDir;		/* 从父路径中分离最后一个目录。 */
	bool		fc_isDbDir = false;	/* 这个目录包含关系吗？ */

	/*
	 * 确定当前路径是否是可以包含关系的数据库目录。
	 *
	 * 首先找到父路径和当前路径之间的分隔符的位置。
	 */
	fc_lastDir = last_dir_separator(fc_path);

	/* 这个路径看起来像是数据库路径吗（即全是数字）？ */
	if (fc_lastDir != NULL &&
		strspn(fc_lastDir + 1, "0123456789") == strlen(fc_lastDir + 1))
	{
		/* 包含父目录的路径部分。 */
		int			fc_parentPathLen = fc_lastDir - fc_path;

		/*
		 * 如果父路径是 $PGDATA/base 或者是表空间版本路径，
		 * 则将路径标记为数据库目录。
		 */
		if (strncmp(fc_path, "./base", fc_parentPathLen) == 0 ||
			(fc_parentPathLen >= (sizeof(TABLESPACE_VERSION_DIRECTORY) - 1) &&
			 strncmp(fc_lastDir - (sizeof(TABLESPACE_VERSION_DIRECTORY) - 1),
					 TABLESPACE_VERSION_DIRECTORY,
					 sizeof(TABLESPACE_VERSION_DIRECTORY) - 1) == 0))
			fc_isDbDir = true;
	}

	fc_dir = AllocateDir(fc_path);
	while ((fc_de = ReadDir(fc_dir, fc_path)) != NULL)
	{
		int			fc_excludeIdx;
		bool		fc_excludeFound;
		ForkNumber	fc_relForkNum; /* 如果文件是关系，则为的分叉类型 */
		int			fc_relOidChars;	/* 文件名中的字符是 rel oid */

		/* 跳过特殊内容 */
		if (strcmp(fc_de->d_name, ".") == 0 || strcmp(fc_de->d_name, "..") == 0)
			continue;

		/* 跳过临时文件 */
		if (strncmp(fc_de->d_name,
					PG_TEMP_FILE_PREFIX,
					strlen(PG_TEMP_FILE_PREFIX)) == 0)
			continue;

		/* 跳过 macOS 系统文件 */
		if (strcmp(fc_de->d_name, ".DS_Store") == 0)
			continue;

		/*
		 * 检查主进程是否已向我们发出退出信号，并在这种情况下
		 * 中止并报错。上面的错误处理程序将为我们调用
		 * do_pg_abort_backup()。还要检查如果备份在恢复时
		 * 开始，服务器没有被提升。
		 * do_pg_backup_stop() 也会检查这一点，但最好是提前
		 * 停止备份，而不是继续到最后并在那里失败。
		 */
		CHECK_FOR_INTERRUPTS();
		if (RecoveryInProgress() != backup_started_in_recovery)
			ereport(ERROR,
					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
					 errmsg("the standby was promoted during online backup"),
					 errhint("This means that the backup being taken is corrupt "
							 "and should not be used. "
							 "Try taking another online backup.")));

		/* 扫描应排除的文件 */
		fc_excludeFound = false;
		for (fc_excludeIdx = 0; excludeFiles[fc_excludeIdx].name != NULL; fc_excludeIdx++)
		{
			int			fc_cmplen = strlen(excludeFiles[fc_excludeIdx].name);

			if (!excludeFiles[fc_excludeIdx].match_prefix)
				fc_cmplen++;
			if (strncmp(fc_de->d_name, excludeFiles[fc_excludeIdx].name, fc_cmplen) == 0)
			{
				elog(DEBUG1, "file \"%s\" excluded from backup", fc_de->d_name);
				fc_excludeFound = true;
				break;
			}
		}

		if (fc_excludeFound)
			continue;

		/* 排除所有未记录表的分叉，除初始化分叉外 */
		if (fc_isDbDir &&
			parse_filename_for_nontemp_relation(fc_de->d_name, &fc_relOidChars,
												&fc_relForkNum))
		{
			/* 从不排除初始化分叉 */
			if (fc_relForkNum != INIT_FORKNUM)
			{
				char		fc_initForkFile[MAXPGPATH];
				char		fc_relOid[OIDCHARS + 1];

				/*
				 * 如果是其他类型的分叉，请检查是否存在具有相同 OID
				 * 的初始化分叉。如果是，则可以排除该文件。
				 */
				memcpy(fc_relOid, fc_de->d_name, fc_relOidChars);
				fc_relOid[fc_relOidChars] = '\0';
				snprintf(fc_initForkFile, sizeof(fc_initForkFile), "%s/%s_init",
						 fc_path, fc_relOid);

				if (lstat(fc_initForkFile, &fc_statbuf) == 0)
				{
					elog(DEBUG2,
						 "unlogged relation file \"%s\" excluded from backup",
						 fc_de->d_name);

					continue;
				}
			}
		}

		/* 排除临时关系 */
		if (fc_isDbDir && looks_like_temp_rel_name(fc_de->d_name))
		{
			elog(DEBUG2,
				 "temporary relation file \"%s\" excluded from backup",
				 fc_de->d_name);

			continue;
		}

		snprintf(fc_pathbuf, sizeof(fc_pathbuf), "%s/%s", fc_path, fc_de->d_name);

		/* 在此跳过 pg_control，以最后备份它 */
		if (strcmp(fc_pathbuf, "./global/pg_control") == 0)
			continue;

		if (lstat(fc_pathbuf, &fc_statbuf) != 0)
		{
			if (errno != ENOENT)
				ereport(ERROR,
						(errcode_for_file_access(),
						 errmsg("could not stat file or directory \"%s\": %m",
								fc_pathbuf)));

			/* 如果在扫描时文件消失了，这不是错误。 */
			continue;
		}

		/* 扫描应排除内容的目录 */
		fc_excludeFound = false;
		for (fc_excludeIdx = 0; excludeDirContents[fc_excludeIdx] != NULL; fc_excludeIdx++)
		{
			if (strcmp(fc_de->d_name, excludeDirContents[fc_excludeIdx]) == 0)
			{
				elog(DEBUG1, "contents of directory \"%s\" excluded from backup", fc_de->d_name);
				fc_convert_link_to_directory(fc_pathbuf, &fc_statbuf);
				fc_size += fc__tarWriteHeader(fc_sink, fc_pathbuf + fc_basepathlen + 1, NULL,
										&fc_statbuf, fc_sizeonly);
				fc_excludeFound = true;
				break;
			}
		}

		if (fc_excludeFound)
			continue;

		/*
		 * 我们可以跳过 pg_wal，WAL 段需要从
		 * WAL 归档中获取。但仍然将其作为空目录包含，
		 * 以便我们获得正确的权限。
		 */
		if (strcmp(fc_pathbuf, "./pg_wal") == 0)
		{
			/* 如果 pg_wal 是符号链接，仍然将其写为目录 */
			fc_convert_link_to_directory(fc_pathbuf, &fc_statbuf);
			fc_size += fc__tarWriteHeader(fc_sink, fc_pathbuf + fc_basepathlen + 1, NULL,
									&fc_statbuf, fc_sizeonly);

			/*
			 * 还需发送 archive_status 目录（通过巧妙地重用
			 * 上面的 statbuf ...）。
			 */
			fc_size += fc__tarWriteHeader(fc_sink, "./pg_wal/archive_status", NULL,
									&fc_statbuf, fc_sizeonly);

			continue;			/* 不要递归进入 pg_wal */
		}

		/* 仅允许在 pg_tblspc 中使用符号链接 */
		if (strcmp(fc_path, "./pg_tblspc") == 0 && S_ISLNK(fc_statbuf.st_mode))
		{
#if defined(HAVE_READLINK) || defined(WIN32)
			char		fc_linkpath[MAXPGPATH];
			int			fc_rllen;

			fc_rllen = readlink(fc_pathbuf, fc_linkpath, sizeof(fc_linkpath));
			if (fc_rllen < 0)
				ereport(ERROR,
						(errcode_for_file_access(),
						 errmsg("could not read symbolic link \"%s\": %m",
								fc_pathbuf)));
			if (fc_rllen >= sizeof(fc_linkpath))
				ereport(ERROR,
						(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
						 errmsg("symbolic link \"%s\" target is too long",
								fc_pathbuf)));
			fc_linkpath[fc_rllen] = '\0';

			fc_size += fc__tarWriteHeader(fc_sink, fc_pathbuf + fc_basepathlen + 1, fc_linkpath,
									&fc_statbuf, fc_sizeonly);
#else

			/*
			 * 如果平台没有符号链接，就不可能存在表空间 -
			 * 显然是其他人创建了它们。对此发出警告并忽略。
			 */
			ereport(WARNING,
					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
					 errmsg("tablespaces are not supported on this platform")));
			continue;
#endif							/* HAVE_READLINK */
		}
		else if (S_ISDIR(fc_statbuf.st_mode))
		{
			bool		fc_skip_this_dir = false;
			ListCell   *fc_lc;

			/*
			 * 在 tar 文件中存储一个目录项，以便我们能够获得
			 * 正确的权限。
			 */
			fc_size += fc__tarWriteHeader(fc_sink, fc_pathbuf + fc_basepathlen + 1, NULL, &fc_statbuf,
									fc_sizeonly);

			/*
			 * 递归调用自己处理一个目录，除非它恰好是位于 PGDATA 内部的一个独立表空间。
			 */
			foreach(fc_lc, fc_tablespaces)
			{
				tablespaceinfo *fc_ti = (tablespaceinfo *) lfirst(fc_lc);

				/*
				 * ti->rpath 是 PGDATA 内部的表空间相对路径，或者
				 * 如果表空间已正确定位到其他地方，则为 NULL。
				 *
				 * 在比较时跳过路径缓冲区中的前导 "./"。
				 */
				if (fc_ti->rpath && strcmp(fc_ti->rpath, fc_pathbuf + 2) == 0)
				{
					fc_skip_this_dir = true;
					break;
				}
			}

			/*
			 * 如果不需要，则跳过发送 pg_tblspc 内部的目录。
			 */
			if (strcmp(fc_pathbuf, "./pg_tblspc") == 0 && !fc_sendtblspclinks)
				fc_skip_this_dir = true;

			if (!fc_skip_this_dir)
				fc_size += fc_sendDir(fc_sink, fc_pathbuf, fc_basepathlen, fc_sizeonly, fc_tablespaces,
								fc_sendtblspclinks, fc_manifest, fc_spcoid);
		}
		else if (S_ISREG(fc_statbuf.st_mode))
		{
			bool		fc_sent = false;

			if (!fc_sizeonly)
				fc_sent = fc_sendFile(fc_sink, fc_pathbuf, fc_pathbuf + fc_basepathlen + 1, &fc_statbuf,
								true, fc_isDbDir ? atooid(fc_lastDir + 1) : InvalidOid,
								fc_manifest, fc_spcoid);

			if (fc_sent || fc_sizeonly)
			{
				/* 添加大小。 */
				fc_size += fc_statbuf.st_size;

				/* 填充到 tar 块大小的倍数。 */
				fc_size += tarPaddingBytesRequired(fc_statbuf.st_size);

				/* 文件头的大小。 */
				fc_size += TAR_BLOCK_SIZE;
			}
		}
		else
			ereport(WARNING,
					(errmsg("skipping special file \"%s\"", fc_pathbuf)));
	}
	FreeDir(fc_dir);
	return fc_size;
}

/*
 * 检查文件是否应该进行校验和验证。
 * 我们只在常规表空间（包括全局和默认）中的文件上验证校验和，
 * 而在这些文件中，有一些文件是被明确排除的。
 */
static bool fc_is_checksummed_file(const char *fc_fullpath, const char *fc_filename)
{
	/* 检查文件是否在表空间中 */
	if (strncmp(fc_fullpath, "./global/", 9) == 0 ||
		strncmp(fc_fullpath, "./base/", 7) == 0 ||
		strncmp(fc_fullpath, "/", 1) == 0)
	{
		int			fc_excludeIdx;

		/* 将文件与 noChecksumFiles 跳过列表进行比较 */
		for (fc_excludeIdx = 0; noChecksumFiles[fc_excludeIdx].name != NULL; fc_excludeIdx++)
		{
			int			fc_cmplen = strlen(noChecksumFiles[fc_excludeIdx].name);

			if (!noChecksumFiles[fc_excludeIdx].match_prefix)
				fc_cmplen++;
			if (strncmp(fc_filename, noChecksumFiles[fc_excludeIdx].name,
						fc_cmplen) == 0)
				return false;
		}

		return true;
	}
	else
		return false;
}

/*****
 * 处理 tar 文件格式的函数
 *
 * 从 pg_dump 复制，但修改为与 libpq 一起使用以进行发送
 */


/*
 * 给定成员，写入 TAR 头并发送文件。
 *
 * 如果 'missing_ok' 为真，则在找不到文件时不会抛出错误。
 *
 * 如果 dboid 不是 InvalidOid，则任何检测到的校验和失败
 * 都会报告给累积统计系统。
 *
 * 如果文件成功发送，则返回 true；如果 'missing_ok' 为真，
 * 并且文件不存在，则返回 false。
 */
static bool fc_sendFile(bbsink *fc_sink, const char *fc_readfilename, const char *fc_tarfilename,
		 struct stat *fc_statbuf, bool fc_missing_ok, Oid fc_dboid,
		 backup_manifest_info *fc_manifest, const char *fc_spcoid)
{
	int			fc_fd;
	BlockNumber fc_blkno = 0;
	bool		fc_block_retry = false;
	uint16		fc_checksum;
	int			fc_checksum_failures = 0;
	off_t		fc_cnt;
	int			fc_i;
	pgoff_t		fc_len = 0;
	char	   *fc_page;
	PageHeader	fc_phdr;
	int			fc_segmentno = 0;
	char	   *fc_segmentpath;
	bool		fc_verify_checksum = false;
	pg_checksum_context fc_checksum_ctx;

	if (pg_checksum_init(&fc_checksum_ctx, fc_manifest->checksum_type) < 0)
		elog(ERROR, "could not initialize checksum of file \"%s\"",
			 fc_readfilename);

	fc_fd = OpenTransientFile(fc_readfilename, O_RDONLY | PG_BINARY);
	if (fc_fd < 0)
	{
		if (errno == ENOENT && fc_missing_ok)
			return false;
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not open file \"%s\": %m", fc_readfilename)));
	}

	fc__tarWriteHeader(fc_sink, fc_tarfilename, NULL, fc_statbuf, false);

	if (!noverify_checksums && DataChecksumsEnabled())
	{
		char	   *fc_filename;

		/*
		 * 获取文件名（不包括路径）。由于 last_dir_separator()
		 * 包括最后的目录分隔符，因此我们通过增加指针来去掉它。
		 */
		fc_filename = last_dir_separator(fc_readfilename) + 1;

		if (fc_is_checksummed_file(fc_readfilename, fc_filename))
		{
			fc_verify_checksum = true;

			/*
			 * 在段边界（"."）处截断，以获取段号
			 * 以便将其与校验和混合。
			 */
			fc_segmentpath = strstr(fc_filename, ".");
			if (fc_segmentpath != NULL)
			{
				fc_segmentno = atoi(fc_segmentpath + 1);
				if (fc_segmentno == 0)
					ereport(ERROR,
							(errmsg("invalid segment number %d in file \"%s\"",
									fc_segmentno, fc_filename)));
			}
		}
	}

	/*
	 * 循环直到我们读取调用者告诉我们预期的数据量。
	 * 如果在我们发送时文件被扩展，文件可以更长，但
	 * 对于基础备份，我们可以忽略这种扩展数据。它将从 WAL 恢复。
	 */
	while (fc_len < fc_statbuf->st_size)
	{
		size_t		fc_remaining = fc_statbuf->st_size - fc_len;

		/* 尝试读取更多数据。 */
		fc_cnt = fc_basebackup_read_file(fc_fd, fc_sink->bbs_buffer,
								   Min(fc_sink->bbs_buffer_length, fc_remaining),
								   fc_len, fc_readfilename, true);

		/*
		 * 校验和在块级别验证，因此我们按块大小 BLCKSZ 遍历
		 * 缓冲区，在确保 TAR_SEND_SIZE/buf 能被 BLCKSZ 整除并且
		 * 我们读取的字节数是 BLCKSZ 的倍数后。
		 */
		Assert((fc_sink->bbs_buffer_length % BLCKSZ) == 0);

		if (fc_verify_checksum && (fc_cnt % BLCKSZ != 0))
		{
			ereport(WARNING,
					(errmsg("could not verify checksum in file \"%s\", block "
							"%u: read buffer size %d and page size %d "
							"differ",
							fc_readfilename, fc_blkno, (int) fc_cnt, BLCKSZ)));
			fc_verify_checksum = false;
		}

		if (fc_verify_checksum)
		{
			for (fc_i = 0; fc_i < fc_cnt / BLCKSZ; fc_i++)
			{
				fc_page = fc_sink->bbs_buffer + BLCKSZ * fc_i;

				/*
				 * 只检查自基础备份开始以来没有被修改的页面。
				 * 否则，它们可能只被写入了一半，校验和将不有效。
				 * 但是，重放 WAL 将在这种情况下恢复正确的页面。
				 * 我们还完全跳过全新的页面，因为它们尚未
				 * 具有校验和。
				 */
				if (!PageIsNew(fc_page) && PageGetLSN(fc_page) < fc_sink->bbs_state->startptr)
				{
					fc_checksum = pg_checksum_page((char *) fc_page, fc_blkno + fc_segmentno * RELSEG_SIZE);
					fc_phdr = (PageHeader) fc_page;
					if (fc_phdr->pd_checksum != fc_checksum)
					{
						/*
						 * 在第一次失败时重试该块。可能我们在 postgres 更新整个块之前读取了块的前4K页面，因此对我们来说它看起来是损坏的。我们只需要重试一次，因为 LSN 应该会更新为我们在下次通过时可以忽略的值。如果错误再次发生，那么这确实是一个验证失败。
						 */
						if (fc_block_retry == false)
						{
							int			fc_reread_cnt;

							/* 重新读取失败的块 */
							fc_reread_cnt =
								fc_basebackup_read_file(fc_fd,
													 fc_sink->bbs_buffer + BLCKSZ * fc_i,
													 BLCKSZ, fc_len + BLCKSZ * fc_i,
													 fc_readfilename,
													 false);
							if (fc_reread_cnt == 0)
							{
								/*
								 * 如果我们遇到文件结尾，必须发生了并发截断，因此就像最初的 fread() 返回 0 一样跳出这个循环。我们将继续处理该情况的相同代码。（不过，我们必须先修复 cnt。）
								 */
								fc_cnt = BLCKSZ * fc_i;
								break;
							}

							/* 设置标志以便我们知道尝试了重试 */
							fc_block_retry = true;

							/* 重置循环以再次验证块 */
							fc_i--;
							continue;
						}

						fc_checksum_failures++;

						if (fc_checksum_failures <= 5)
							ereport(WARNING,
									(errmsg("checksum verification failed in "
											"file \"%s\", block %u: calculated "
											"%X but expected %X",
											fc_readfilename, fc_blkno, fc_checksum,
											fc_phdr->pd_checksum)));
						if (fc_checksum_failures == 5)
							ereport(WARNING,
									(errmsg("further checksum verification "
											"failures in file \"%s\" will not "
											"be reported", fc_readfilename)));
					}
				}
				fc_block_retry = false;
				fc_blkno++;
			}
		}

		/*
		 * 如果我们遇到文件结尾，必须发生了并发截断。这不是错误条件，因为 WAL 重放将会修复问题。
		 */
		if (fc_cnt == 0)
			break;

		/* 归档我们刚刚读取的数据。 */
		bbsink_archive_contents(fc_sink, fc_cnt);

		/* 还将其提供给校验和机制。 */
		if (pg_checksum_update(&fc_checksum_ctx,
							   (uint8 *) fc_sink->bbs_buffer, fc_cnt) < 0)
			elog(ERROR, "could not update checksum of base backup");

		fc_len += fc_cnt;
	}

	/* 如果在我们发送时文件被截断，则用零填充它 */
	while (fc_len < fc_statbuf->st_size)
	{
		size_t		fc_remaining = fc_statbuf->st_size - fc_len;
		size_t		fc_nbytes = Min(fc_sink->bbs_buffer_length, fc_remaining);

		MemSet(fc_sink->bbs_buffer, 0, fc_nbytes);
		if (pg_checksum_update(&fc_checksum_ctx,
							   (uint8 *) fc_sink->bbs_buffer,
							   fc_nbytes) < 0)
			elog(ERROR, "could not update checksum of base backup");
		bbsink_archive_contents(fc_sink, fc_nbytes);
		fc_len += fc_nbytes;
	}

	/*
	 * 根据 tar 格式要求填充到块边界。（这一小段数据可能不值得节流，并且未进行校验和处理，因为它实际上不是文件的一部分。）
	 */
	fc__tarWritePadding(fc_sink, fc_len);

	CloseTransientFile(fc_fd);

	if (fc_checksum_failures > 1)
	{
		ereport(WARNING,
				(errmsg_plural("file \"%s\" has a total of %d checksum verification failure",
							   "file \"%s\" has a total of %d checksum verification failures",
							   fc_checksum_failures,
							   fc_readfilename, fc_checksum_failures)));

		pgstat_report_checksum_failures_in_db(fc_dboid, fc_checksum_failures);
	}

	total_checksum_failures += fc_checksum_failures;

	AddFileToBackupManifest(fc_manifest, fc_spcoid, fc_tarfilename, fc_statbuf->st_size,
							(pg_time_t) fc_statbuf->st_mtime, &fc_checksum_ctx);

	return true;
}

static int64 fc__tarWriteHeader(bbsink *fc_sink, const char *fc_filename, const char *fc_linktarget,
				struct stat *fc_statbuf, bool fc_sizeonly)
{
	enum tarError fc_rc;

	if (!fc_sizeonly)
	{
		/*
		 * 截至目前，支持的最小块大小为 1kB，是 TAR_BLOCK_SIZE 的两倍。由于缓冲区大小必须是 BLCKSZ 的倍数，因此可以安全地假设缓冲区足够大，可以容纳整个 tar 块。我们通过这些断言进行二次确认。
		 */
		StaticAssertStmt(TAR_BLOCK_SIZE <= BLCKSZ,
						 "BLCKSZ too small for tar block");
		Assert(fc_sink->bbs_buffer_length >= TAR_BLOCK_SIZE);

		fc_rc = tarCreateHeader(fc_sink->bbs_buffer, fc_filename, fc_linktarget,
							 fc_statbuf->st_size, fc_statbuf->st_mode,
							 fc_statbuf->st_uid, fc_statbuf->st_gid,
							 fc_statbuf->st_mtime);

		switch (fc_rc)
		{
			case TAR_OK:
				break;
			case TAR_NAME_TOO_LONG:
				ereport(ERROR,
						(errmsg("file name too long for tar format: \"%s\"",
								fc_filename)));
				break;
			case TAR_SYMLINK_TOO_LONG:
				ereport(ERROR,
						(errmsg("symbolic link target too long for tar format: "
								"file name \"%s\", target \"%s\"",
								fc_filename, fc_linktarget)));
				break;
			default:
				elog(ERROR, "unrecognized tar error: %d", fc_rc);
		}

		bbsink_archive_contents(fc_sink, TAR_BLOCK_SIZE);
	}

	return TAR_BLOCK_SIZE;
}

/*
 * 使用零字节填充到 TAR_BLOCK_SIZE 的倍数。
 */
static void fc__tarWritePadding(bbsink *fc_sink, int fc_len)
{
	int			fc_pad = tarPaddingBytesRequired(fc_len);

	/*
	 * 在 _tarWriteHeader 中，应该可以安全地假设缓冲区足够大，因此我们不需要将其分多次进行。
	 */
	Assert(fc_sink->bbs_buffer_length >= TAR_BLOCK_SIZE);
	Assert(fc_pad <= TAR_BLOCK_SIZE);

	if (fc_pad > 0)
	{
		MemSet(fc_sink->bbs_buffer, 0, fc_pad);
		bbsink_archive_contents(fc_sink, fc_pad);
	}
}

/*
 * 如果 statbuf 中的条目是链接，则调整 statbuf 使其看起来像目录，以便以这种方式写入。
 */
static void fc_convert_link_to_directory(const char *fc_pathbuf, struct stat *fc_statbuf)
{
	/* 如果是符号链接，仍然将其作为目录写入 */
	if (S_ISLNK(fc_statbuf->st_mode))
		fc_statbuf->st_mode = S_IFDIR | pg_dir_create_mode;
}

/*
 * 从文件中读取一些数据，设置等待事件并报告遇到的任何错误。
 *
 * 如果 partial_read_ok 为 false，如果读取的字节数不等于请求的字节数，也报告错误。
 *
 * 返回读取的字节数。
 */
static int fc_basebackup_read_file(int fc_fd, char *fc_buf, size_t fc_nbytes, off_t fc_offset,
					 const char *fc_filename, bool fc_partial_read_ok)
{
	int			fc_rc;

	pgstat_report_wait_start(WAIT_EVENT_BASEBACKUP_READ);
	fc_rc = pg_pread(fc_fd, fc_buf, fc_nbytes, fc_offset);
	pgstat_report_wait_end();

	if (fc_rc < 0)
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not read file \"%s\": %m", fc_filename)));
	if (!fc_partial_read_ok && fc_rc > 0 && fc_rc != fc_nbytes)
		ereport(ERROR,
				(errcode_for_file_access(),
				 errmsg("could not read file \"%s\": read %d of %zu",
						fc_filename, fc_rc, fc_nbytes)));

	return fc_rc;
}
